package com.qyer.commons.concurrent.assemblingline;

import static org.apache.commons.io.IOUtils.closeQuietly;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.collect.Lists;
import com.qyer.commons.concurrent.ArrayListBulk;
import com.qyer.commons.concurrent.GenericLinkedBlockingQueue;
import com.qyer.commons.mc.converter.Converter;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.DecimalFormat;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * User: Z J Wu Date: 2016/06/04 Time: 17:09 Package: com.qyer.commons.concurrent
 */
public class AssemblingLine<T> extends AbstractAssemblingLine {

  private static final Logger LOGGER = LoggerFactory.getLogger(AssemblingLine.class);

  private ArrayListBulk<T> pillMaker = ArrayListBulk.PILL_MAKER.createPill();
  private FileBulkRawDataGenerator<T> rawDataGenerator;
  private Summarizer<T> summarizer;

  private List<AssemblerGroup<T>> assemblerGroups = Lists.newArrayListWithExpectedSize(3);

  private StatisticContext context = new StatisticContext();

  private CountDownLatch signal = new CountDownLatch(1);

  private int maxEntitiesInRawDataConvey;

  public AssemblingLine(StatisticContext context) {
    this.context = context;
  }

  public AssemblingLine(StatisticContext context, int maxEntitiesInRawDataConvey) {
    this.context = context;
    this.maxEntitiesInRawDataConvey = maxEntitiesInRawDataConvey;
  }

  public AssemblingLine<T> setFileDataGenerator(String name, String inputPath,
                                                Converter<String, T> converter) {
    this.rawDataGenerator = new FileBulkRawDataGenerator<>(name, this.context, inputPath,
                                                           converter);
    return this;
  }

  public AssemblingLine<T> setFileSummarizer(String name,
                                             AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>>
                                               h) {
    this.summarizer = new Summarizer<>(name, this.signal, h, context);
    return this;
  }

  public AssemblingLine<T> addAssemblerGroup(String name, int assemblerCount,
                                             AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>> h,
                                             int acceptBulkSize) {
    DecimalFormat df = new DecimalFormat("000");
    AssemblerGroup<T> assemblerGroup = new AssemblerGroup<>(name, assemblerCount, acceptBulkSize);
    for (int i = 0; i < assemblerCount; i++) {
      String n = name + df.format((i + 1));
      assemblerGroup.addAssembler(new BulkAssembler<>(name, n, h, context));
    }
    assemblerGroups.add(assemblerGroup);
    return this;
  }

  public void connect() throws AssemblingLineException {
    if (rawDataGenerator == null || summarizer == null || CollectionUtils
      .isEmpty(assemblerGroups)) {
      throw new AssemblingLineException(
        "Must deploy at least one raw data generator, one summarizer and one group of " +
          "assemblers.");
    }
    GenericLinkedBlockingQueue<ArrayListBulk<T>> c1, c2;
    Notifier<ArrayListBulk<T>> notifier;
    Differential<T> differential;

    AssemblerGroup<T> currentGroup = assemblerGroups.get(0), nextGroup;
    String n = rawDataGenerator.getName() + CS + currentGroup.getName();
    if (maxEntitiesInRawDataConvey <= 0) {
      c1 = new GenericLinkedBlockingQueue<>(n);
    } else {
      c1 = new GenericLinkedBlockingQueue<>(n, maxEntitiesInRawDataConvey);
    }

    notifier = Notifier.newCommonNotifier(1, c1, currentGroup.size(), pillMaker);
    rawDataGenerator.assembleConveyors(c1);
    rawDataGenerator.assembleNotifier(notifier);
    rawDataGenerator.setBulkSize(currentGroup.getAcceptBulkSize());
    boolean needDifferential = assemblerGroups.size() > 1;

    for (int i = 0; i < assemblerGroups.size() - 1; i++) {
      currentGroup = assemblerGroups.get(i);
      nextGroup = assemblerGroups.get(i + 1);
      c2 = new GenericLinkedBlockingQueue<>(currentGroup.getName() + CS + nextGroup.getName());
      notifier = Notifier.newCommonNotifier(currentGroup.size(), c2, nextGroup.size(), pillMaker);
      for (BulkAssembler<T> assembler : currentGroup.getAssemblers()) {
        assembler.assembleConveyors(c1, c2);
        assembler.assembleNotifier(notifier);
        if (needDifferential) {
          differential = new Differential<>(nextGroup.getAcceptBulkSize(), c2);
          assembler.setDifferential(differential);
        }
      }
      c1 = c2;
    }
    currentGroup = assemblerGroups.get(assemblerGroups.size() - 1);
    c2 = new GenericLinkedBlockingQueue<>(currentGroup.getName() + CS + summarizer.getName());
    notifier = new Notifier<>(currentGroup.size(), c2, 1, pillMaker);
    for (BulkAssembler<T> assembler : currentGroup.getAssemblers()) {
      assembler.assembleConveyors(c1, c2);
      assembler.assembleNotifier(notifier);
    }
    summarizer.assembleConveyors(c2);
  }

  public StatisticContext start() throws AssemblingLineException, JsonProcessingException {
    long t1 = System.currentTimeMillis();
    try {
      // 1. Start summarizer
      new Thread(summarizer, summarizer.getName()).start();
      // 2. Start assemblers
      for (AssemblerGroup<T> assemblerGroup : assemblerGroups) {
        for (int i = 0; i < assemblerGroup.size(); i++) {
          BulkAssembler<T> assembler = assemblerGroup.getAssembler(i);
          new Thread(assembler, assembler.getName()).start();
        }
      }
      // 3. Start raw data generator in this thread.
      rawDataGenerator.beginJob();
      signal.await();
      return context;
    } catch (Exception e) {
      throw new AssemblingLineException("Assembling line get an error.", e);
    } finally {
      long t2 = System.currentTimeMillis();
      closeQuietly(rawDataGenerator);
      closeQuietly(summarizer);
      LOGGER.info("Assembling line finished job in {} milliseconds.", (t2 - t1));
    }
  }

  public ArrayListBulk<T> getPillMaker() {
    return pillMaker;
  }

  public FileBulkRawDataGenerator<T> getRawDataGenerator() {
    return rawDataGenerator;
  }

  public Summarizer<T> getSummarizer() {
    return summarizer;
  }

  public List<AssemblerGroup<T>> getAssemblerGroups() {
    return assemblerGroups;
  }

  public StatisticContext getContext() {
    return context;
  }

  public CountDownLatch getSignal() {
    return signal;
  }
}
